package com.demo.top;

import cn.hutool.json.JSONUtil;
import com.demo.client.RedisClient;
import com.demo.domain.EventEntity;
import com.demo.domain.RealtimeEventEntity;
import com.demo.domain.TopProductEntity;
import com.demo.util.DateUtil;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Objects;

/**
 * @author hornsey
 * @date 2022/10/19
 */
public class ActiveProcess
		extends ProcessWindowFunction<EventEntity, RealtimeEventEntity, Tuple, TimeWindow> {

	RedisClient redisClient = new RedisClient();

	@Override
	public void process(Tuple key, Context context, Iterable<EventEntity> iterable,
	                    Collector<RealtimeEventEntity> collector) throws Exception {
		long count = 0;
		int itemId = key.getField(0);
		String activeType = key.getField(1);
		System.out.println("ActiveProcess key:" + JSONUtil.toJsonStr(key));
		for (EventEntity entity : iterable) {
//			System.out.println("ActiveProcess entity:" + JSONUtil.toJsonStr(entity));
			count = count + 1;
		}
		TimeWindow window = context.window();
		RealtimeEventEntity rtEntry = RealtimeEventEntity.of(itemId, activeType, window.getEnd(), count);
		save(rtEntry);
		collector.collect(rtEntry);
	}

	@Override
	public void clear(Context context) throws Exception {
		super.clear(context);
	}

	private void save(RealtimeEventEntity entity) {
		if (Objects.isNull(redisClient)) {
			System.out.println("redisClient is null");
		}

		long expireAtTime = DateUtil.getCurrentDate()/1000 + 86400 - 10;

		String key = "rt" + entity.getUserId();
		String tagName;
		if (entity.getAppActive() == 1) {
			tagName = "isAppActive";
		} else if (entity.getMiniProgramActive() == 1) {
			tagName = "isMiniActive";
		} else {
			return;
		}

		redisClient.setHashData(key, tagName, "1", expireAtTime);
		System.out.println("save success " + JSONUtil.toJsonStr(entity));
	}
}
